package org.sxp.datax.executor.service.jobhandler;

import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.PPJobTmpParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobContext;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.core.thread.TriggerCallbackThread;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.sxp.common.dto.DatabaseDTO;
import org.sxp.common.dto.DevelopJobDTO;
import org.sxp.common.dto.JobParamDTO;
import org.sxp.common.enums.DbTypeEnum;
import org.sxp.common.enums.JobTypeEnum;
import org.sxp.datax.executor.constants.DataXConstant;
import org.sxp.datax.executor.service.command.BuildCommand;
import org.sxp.datax.executor.service.logparse.AnalysisCount;
import org.sxp.datax.executor.service.logparse.LogStatistics;
import org.sxp.datax.executor.util.DbUtil;
import org.sxp.datax.executor.util.ProcessUtil;
import org.sxp.datax.executor.util.SystemUtils;

import java.io.*;
import java.util.List;
import java.util.concurrent.FutureTask;

/**
 * DataX任务运行
 * @author shenxingping
 */
@Component
public class ExecutorJobHandler {

    public static final String SELECT = "select";
    @Value("${datax.jsonpath:}")
    private String jsonPath;

    @Value("${datax.pypath:}")
    private String dataXPyPath;


    @XxlJob("executorJobHandler")
    public void execute() {
        if(IJobHandler.jobTmpFiles.get(XxlJobHelper.getJobId()) != null){
            XxlJobHelper.log("当前执行器存在此调度任务，勿重复执行!!!");
            return;
        }
        String params = XxlJobHelper.getJobParam();
        JobParamDTO jobDTO = JSON.parseObject(params, JobParamDTO.class);
        if(jobDTO != null && jobDTO.getType() != null && jobDTO.getParam() != null){
            if(jobDTO.getType().intValue() == JobTypeEnum.sync.getCode().intValue()){
                syncJob(jobDTO.getParam());
            }else if(jobDTO.getType().intValue() == JobTypeEnum.develop.getCode().intValue()){
                developJob(jobDTO.getParam());
            }else if(jobDTO.getType().intValue() == JobTypeEnum.flow.getCode().intValue()){
                flowJob(jobDTO.getParam());
            }else{
                XxlJobHelper.log("不支持的调度类型：type={}", jobDTO.getType());
            }
        }else{
            XxlJobHelper.log("调度参数异常：params={}", params);
        }
    }

    private void flowJob(String params){
        XxlJobHelper.log("**************************************************执行【作业流程】开始**************************************************");
        if(StrUtil.isBlank(params)){
            XxlJobHelper.log("脚本参数为空!");
            return;
        }
        try{
            List<JobParamDTO> jobParamDTOList = JSON.parseArray(params, JobParamDTO.class);
            for (JobParamDTO paramDTO : jobParamDTOList){
                if(paramDTO.getType().intValue() == JobTypeEnum.sync.getCode().intValue()){
                    syncJob(paramDTO.getParam());
                }else if(paramDTO.getType().intValue() == JobTypeEnum.develop.getCode().intValue()){
                    developJob(paramDTO.getParam());
                }else{
                    XxlJobHelper.log("不支持的调度类型：type={}", paramDTO.getType());
                }
            }
        }catch (Exception e){
            XxlJobHelper.log(e);
        }
        XxlJobHelper.log("**************************************************执行【作业流程】结束**************************************************");
    }

    private void developJob(String params){
        XxlJobHelper.log("★★★★★★★★★★★★★★★★★★★执行【数据开发】开始★★★★★★★★★★★★★★★★★★★");
        if(StrUtil.isBlank(params)){
            XxlJobHelper.log("脚本参数为空!");
            return;
        }
        long beginTime = System.currentTimeMillis();
        DevelopJobDTO jobDTO = JSON.parseObject(params, DevelopJobDTO.class);
        try{
            IJobHandler.jobTmpFiles.put(String.valueOf(XxlJobHelper.getJobId()), new PPJobTmpParam());
            if(jobDTO.getSqlScript().toLowerCase().startsWith(SELECT)){
                XxlJobHelper.log("select 查询语句无须同步，任务结束!!!");
            }else{
                DatabaseDTO databaseDTO = new DatabaseDTO();
                databaseDTO.setType(DbTypeEnum.mysql.getCode());
                databaseDTO.setUrl(jobDTO.getJdbcUrl());
                databaseDTO.setUsername(jobDTO.getUsername());
                databaseDTO.setPassword(jobDTO.getPassword());
                int index = DbUtil.excuteCUD(databaseDTO, jobDTO.getSqlScript());
                XxlJobHelper.log("受影响行数：{}", index);
            }
        }catch (Exception ex){
            XxlJobHelper.log(ex);
        }finally {
            long endTime = System.currentTimeMillis();
            XxlJobHelper.log("任务结束，耗时：{}秒", (endTime-beginTime)/1000);
        }
        XxlJobHelper.log("★★★★★★★★★★★★★★★★★★★执行【数据开发】结束★★★★★★★★★★★★★★★★★★★");
    }


    private void syncJob(String params){
        XxlJobHelper.log("★★★★★★★★★★★★★★★★★★★执行【离线同步】开始★★★★★★★★★★★★★★★★★★★");
        if(StrUtil.isBlank(params)){
            XxlJobHelper.log("脚本参数为空!");
            return;
        }
        int exitValue = -1;
        Thread errThread = null;
        String tmpFilePath;
        tmpFilePath = generateTemJsonFile(params);

        try {
            String[] cmdarrayFinal = BuildCommand.buildDataXExecutorCmd(tmpFilePath,dataXPyPath);
            final Process process = Runtime.getRuntime().exec(cmdarrayFinal);
            String prcsId = ProcessUtil.getProcessId(process);
            XxlJobHelper.log("------------------DataX process id: " + prcsId);
            PPJobTmpParam jobTmpParam = new PPJobTmpParam();
            jobTmpParam.setProcessId(prcsId);
            jobTmpParam.setJobTmpFile(tmpFilePath);
            IJobHandler.jobTmpFiles.put(String.valueOf(XxlJobHelper.getJobId()), jobTmpParam);
            XxlJobHelper.log("DataX已启动，启动进程ID={}", prcsId);
            // log-thread
            Thread futureThread = null;
            FutureTask<LogStatistics> futureTask = new FutureTask<LogStatistics>(() -> AnalysisCount.analysisLog(new BufferedInputStream(process.getInputStream())));
            futureThread = new Thread(futureTask);
            futureThread.start();

            errThread = new Thread(() -> {
                try {
                    AnalysisCount.analysisLog(new BufferedInputStream(process.getErrorStream()));
                } catch (IOException e) {
                    XxlJobHelper.log(e);
                }
            });

            errThread.start();
            // process-wait
            // exit code: 0=success, 1=error
            exitValue = process.waitFor();
            // log-thread join
            errThread.join();
        } catch (InterruptedException e){
            XxlJobHelper.log("DataX进程终止退出！", e);
        }catch (Exception e) {
            XxlJobHelper.log(e);
        } finally {
            if (errThread != null && errThread.isAlive()) {
                errThread.interrupt();
            }
            IJobHandler.jobTmpFiles.remove(XxlJobHelper.getJobId());
            //  删除临时文件
            if (FileUtil.exist(tmpFilePath)) {
                FileUtil.del(new File(tmpFilePath));
            }
        }
        if (exitValue == 0) {
            XxlJobHelper.log("DataX进程结束，原因：正常结束!!!");
        } else {
            XxlJobHelper.log("DataX进程结束，原因：进程终止!!!");
        }
        try{
            Thread.sleep(5000);
        }catch (InterruptedException e){
            XxlJobHelper.log("调度执行结束!!!");
        }
        XxlJobHelper.log("★★★★★★★★★★★★★★★★★★★执行【离线同步】结束★★★★★★★★★★★★★★★★★★★");
    }

    private String generateTemJsonFile(String jobJson) {
        String tmpFilePath;
        String dataXHomePath = SystemUtils.getDataXHomePath();
        if (StringUtils.isNotEmpty(dataXHomePath)) {
            jsonPath = dataXHomePath + DataXConstant.DEFAULT_JSON;
        }
        if (!FileUtil.exist(jsonPath)) {
            FileUtil.mkdir(jsonPath);
        }
        tmpFilePath = jsonPath + File.separator + "jobTmp-" + IdUtil.simpleUUID() + ".conf";
        // 根据json写入到临时本地文件
        try (PrintWriter writer = new PrintWriter(tmpFilePath, "UTF-8")) {
            writer.println(jobJson);
        } catch (FileNotFoundException | UnsupportedEncodingException e) {
            XxlJobHelper.log("JSON 临时文件写入异常：" + e.getMessage());
        }
        return tmpFilePath;
    }

    public void init(){
        XxlJobHelper.log("init");
    }
    public void destroy(){
        XxlJobHelper.log("destory");
    }

}
